package com.hkbigdata.streamCoreCoding;


import com.hkbigdata.entry.UserBehavior;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @author liuanbo
 * @creat 2023-04-10-13:01
 * @see 2194550857@qq.com
 */
public class Flink01_PageView_WordCount {
    public static void main(String[] args) throws Exception {
        //1.运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取数据
        DataStreamSource<String> source = env.readTextFile("input/UserBehavior.csv");

        //3.拍平数据并且转换为Tuple2()
        SingleOutputStreamOperator<Tuple2<String, Integer>> StringToTuple = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] split = value.split(",");

                //4.使用javabean来接受数据
                UserBehavior userBehavior = new UserBehavior(
                        Long.parseLong(split[0]),
                        Long.parseLong(split[1]),
                        Integer.parseInt(split[2]),
                        split[3],
                        Long.parseLong(split[4])
                );

                //5.判断userBehavior的behavior值是否为pv，如果是那么就将其转换为Tuple2
                if ("pv".equals(userBehavior.getBehavior())) {
                    out.collect(new Tuple2<>("pv", 1));
                }

            }
        });

        //6.分组聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = StringToTuple.keyBy(data -> data.f0).sum(1);

        //7.打印
        sum.print();

        //8.执行
        env.execute();

    }
}
